package com.sunday.common.mq.rabbit.study.spring.study.e3_Receiving_Messages;


import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;
import java.util.Map;

/**
 * org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration
 * https://github.com/openzipkin/brave/blob/master/instrumentation/spring-rabbit/README.md
 */
@Slf4j
@Configuration
public class RabbitConfiguration implements InitializingBean {

    @Autowired
    private CachingConnectionFactory rabbitConnectionFactory;

    @Value("${spring.application.name}")
    private String applicationName;

    public static final String EXCHANGE_NAME = "receiving_messages_exchange";
    public static final String ROUTING_KEY_NAME = "receiving_messages_routing_key";
    public static final String QUEUE_NAME = "receiving_messages_queue";

    /**
     * 自动声明交换、队列和绑定
     * 该组件可以在启动时声明交换、队列和绑定。 它懒惰地通过 . 因此，如果启动时代理不存在，则无关紧要。 第一次使用 a 时（例如， 通过发送消息）侦听器触发并应用管理功能。
     * 在侦听器中执行自动声明的另一个好处是，如果连接因任何原因而断开（例如， 代理死亡、网络故障等），当重新建立连接时，将再次应用它们。RabbitAdminConnectionListenerConnection
     * <p>
     * 以这种方式声明的队列必须具有固定的名称 — 显式声明或由实例框架生成。 匿名队列是非持久队列、独占队列和自动删除队列。AnonymousQueue
     * 仅当缓存模式为 （默认值） 时，才会执行自动声明。 存在此限制的原因是独占队列和自动删除队列绑定到连接。CachingConnectionFactory CHANNEL
     */
    @Bean
    public Queue queue() {
        return new Queue(QUEUE_NAME);
    }

    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(EXCHANGE_NAME);
    }

    @Bean
    public Binding binding() {
        return new Binding(QUEUE_NAME, Binding.DestinationType.QUEUE, EXCHANGE_NAME, ROUTING_KEY_NAME, null);
    }

    @Override
    public void afterPropertiesSet() {
        rabbitConnectionFactory.setConnectionNameStrategy(connectionFactory -> applicationName);
    }


    /**
     * 容器
     * 现在您已经了解了 -listen 回调的各种选项，我们可以将注意力转向容器。
     * 基本上，容器处理“主动”职责，以便侦听器回调可以保持被动。 容器是“生命周期”组件的一个示例。 它提供了启动和停止的方法。
     * 配置容器时，实质上是弥合 AMQP 队列和实例之间的差距。 您必须提供对 和队列名称或队列实例的引用，该侦听器应从中使用这些消息。
     *
     * https://docs.spring.io/spring-amqp/docs/current/reference/html/#simplemessagelistenercontainer
     *
     * MessageMessageListener ConnectionFactor
     * 默认情况下，侦听器容器启动从队列接收消息的单个使用者。
     * 检查上一节中的表时，可以看到许多控制并发的属性。 最简单的是 ，它创建并发处理消息的（固定）数量的使用者。
     * concurrentConsumers
     * 在版本 1.3.0 之前，这是唯一可用的设置，必须停止并重新启动容器才能更改设置。
     * 从版本 1.3.0 开始，现在可以动态调整属性。 如果在容器运行时更改了使用者，则会根据需要添加或删除使用者，以调整到新设置。
     * concurrentConsumers
     * 此外，还添加了一个名为的新属性，容器会根据工作负载动态调整并发性。 这将与四个附加属性结合使用：、、 和 。 使用默认设置，增加使用者的算法的工作方式如下：
     * maxConcurrentConsumers consecutiveActiveTrigger startConsumerMinInterval consecutiveIdleTrigger stopConsumerMinInterval
     * 如果未达到 ，并且现有使用者连续十个周期处于活动状态，并且自启动最后一个使用者以来至少已经过去了 10 秒，则启动新的使用者。 如果使用者在 * 毫秒内至少收到一条消息，则认为该使用者处于活动状态。maxConcurrentConsumersbatchSizereceiveTimeout
     * 使用默认设置，减少使用者的算法的工作方式如下：
     * 如果运行次数超过，并且使用者检测到连续十次超时（空闲），并且最后一个使用者至少在 60 秒前停止，则使用者将停止。 超时取决于 和 属性。 如果使用者在 * 毫秒内未收到任何消息，则将其视为空闲。 因此，在默认超时（40 秒）和 <> 秒的情况下，在空闲时间 <> 秒后考虑停止使用者（四个超时对应于一个空闲检测）。
     * concurrentConsumers receiveTimeout batchSize batchSize receiveTimeout batchSize
     *
     * 实际上，只有当整个容器闲置一段时间时，才能停止使用者。 这是因为代理在所有活跃消费者之间共享其工作。
     * 每个使用者使用单个通道，而不考虑配置的队列数量。
     * 从版本 2.0 开始，可以使用属性设置 and 属性，例如 。
     * concurrentConsumers maxConcurrentConsumers concurrency 2-4
     */
//    @Bean
    public SimpleMessageListenerContainer messageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setListenerId(MQApi.listenerId);
        container.setConnectionFactory(rabbitConnectionFactory);
        container.setQueueNames(QUEUE_NAME);
//        container.setMessageListener(exampleListener());
        container.setMessageListener(new ExtendedListenerAdapter(new MyMessageListener()));
        // 消费者优先 从 RabbitMQ V3.2 开始，代理现在支持使用者优先级（请参阅将使用者优先级与 RabbitMQ 配合使用）。 这是通过在使用者上设置参数来实现的。 现在支持设置使用者参数，如以下示例所示：x-prioritySimpleMessageListenerContainer
        container.setConsumerArguments(Map.of("x-priority", Integer.valueOf(10)));
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return container;
    }

    /**
     * 选择容器
     * 版本 2.0 引入了 （DMLC）。 以前，只有 （SMLC） 可用。 SMLC 为每个使用者使用内部队列和专用线程。
     * 如果将容器配置为侦听多个队列，则使用相同的使用者线程来处理所有队列。 并发性由 和其他属性控制。
     * 当消息从 RabbitMQ 客户端到达时，客户端线程通过队列将它们传递给使用者线程。
     * 之所以需要此体系结构，是因为在早期版本的 RabbitMQ 客户端中，无法进行多个并发交付。
     * 较新版本的客户端具有修订的线程模型，现在可以支持并发。 这允许引入DMLC，其中侦听器现在直接在RabbitMQ客户端线程上调用。
     * 因此，它的架构实际上比SMLC“更简单”。 但是，此方法存在一些限制，并且 DMLC 不提供 SMLC 的某些功能。
     * 此外，并发性由 （和客户端库的线程池） 控制。 和关联的属性不适用于此容器。
     * DirectMessageListenerContainer
     * SimpleMessageListenerContainer concurrentConsumers consumersPerQueue concurrentConsumers
     * 以下功能可用于 SMLC，但不适用于 DMLC：
     * batchSize：使用 SMLC，您可以设置此项以控制事务中传递的消息数或减少确认数，但这可能会导致失败后重复传递数增加。 （DMLC 确实有 ，您可以使用它来减少确认，与 with 和 SMLC 相同，但它不能用于事务 — 每条消息都在单独的事务中传递和确认）。messagesPerAckbatchSize
     * consumerBatchEnabled：在使用者中启用离散消息的批处理;有关详细信息，请参阅消息侦听器容器配置。
     * maxConcurrentConsumers和使用者缩放间隔或触发器 — DMLC 中没有自动缩放。 但是，它确实允许您以编程方式更改属性，并相应地调整使用者。consumersPerQueue
     * 但是，DMLC 与 SMLC 相比具有以下优势：
     * 在运行时添加和删除队列更有效。 使用 SMLC，将重新启动整个使用者线程（取消并重新创建所有使用者）。 使用 DMLC，不会取消未受影响的消费者。
     * 避免了 RabbitMQ 客户端线程和使用者线程之间的上下文切换。
     * 线程在使用者之间共享，而不是为 SMLC 中的每个使用者提供专用线程。 但是，请参阅线程和异步使用者中有关连接工厂配置的重要说明。
     * 有关哪些配置属性适用于每个容器的信息，请参阅消息侦听器容器配置。
     *
     * https://docs.spring.io/spring-amqp/docs/current/reference/html/#using-directmessagelistenercontainer
     *
     * DirectMessageListenerContainer
     * 使用此容器，并发性基于配置的队列和 . 每个队列的每个使用者使用单独的通道，并发性由 rabbit 客户端库控制。 默认情况下，在撰写本文时，它使用线程池。consumersPerQueueDEFAULT_NUM_THREADS = Runtime.getRuntime().availableProcessors() * 2
     * 您可以配置 以提供所需的最大并发性。taskExecutor
     *
     */
//    @Bean
    public DirectMessageListenerContainer directMessageListenerContainer() {
        DirectMessageListenerContainer container = new DirectMessageListenerContainer();
        container.setListenerId(MQApi.listenerId);
        container.setConnectionFactory(rabbitConnectionFactory);
        container.setQueueNames(QUEUE_NAME);
//        container.setMessageListener(exampleListener());
        container.setMessageListener(new ExtendedListenerAdapter(new MyMessageListener()));
        // 消费者优先 从 RabbitMQ V3.2 开始，代理现在支持使用者优先级（请参阅将使用者优先级与 RabbitMQ 配合使用）。 这是通过在使用者上设置参数来实现的。 现在支持设置使用者参数，如以下示例所示：x-prioritySimpleMessageListenerContainer
        container.setConsumerArguments(Map.of("x-priority", Integer.valueOf(10)));
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        /**
         * 检测空闲异步使用者
         * 虽然高效，但异步使用者的一个问题是检测它们何时处于空闲状态 — 用户可能希望采用 如果一段时间内没有消息到达，则执行某些操作。
         * 从版本 1.6 开始，现在可以将侦听器容器配置为在一段时间过去而没有消息传递时发布 。 当容器处于空闲状态时，每隔几毫秒发布一个事件。ListenerContainerIdleEventidleEventInterval
         * 若要配置此功能，请在容器上设置。 下面的示例演示如何在 XML 和 Java 中执行此操作（对于 a 和 a）：idleEventIntervalSimpleMessageListenerContainerSimpleRabbitListenerContainerFactory
         */
        container.setIdleEventInterval(60000L);
        return container;
    }


    private final static MessageListener LISTENER = (message) -> log.info("{}", message);

    @Bean
    public MessageListener exampleListener() {
        return new MessageListener() {
            public void onMessage(Message message) {
                log.info("received: " + message);
            }
        };
    }

    public class MyMessageListener {
        public void handleMessage(Object extractedMessage, Channel channel, Message message) throws IOException {
            log.info("extractedMessage: " + extractedMessage);
            log.info("channel: " + channel);
            log.info("message: " + message);
            log.info("tag: " + message.getMessageProperties().getDeliveryTag());
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }

    public class ExtendedListenerAdapter extends MessageListenerAdapter {

        public ExtendedListenerAdapter(Object delegate) {
            super(delegate);
        }

        @Override
        protected Object[] buildListenerArguments(Object extractedMessage, Channel channel, Message message) {
            return new Object[]{extractedMessage, channel, message};
        }

    }

}
